package Others.Canal.service;


import Others.Canal.properties.CanalProperties;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.dao.InvalidDataAccessResourceUsageException;
import org.springframework.jdbc.core.JdbcTemplate;

import java.sql.Statement;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

@Slf4j
public class CanalService {

    @Autowired
    @Qualifier("canalConnector")
    private CanalConnector canalConnector;

    @Autowired
    private CanalProperties canalProperties;

    // 目标数据库的JDBC模板（用于写入同步数据）
    @Autowired
    @Qualifier("targetJdbcTemplate")
    private JdbcTemplate targetJdbcTemplate;

    // 源数据库的JDBC模板（用于全量同步）
    @Autowired
    @Qualifier("sourceJdbcTemplate")
    private JdbcTemplate sourceJdbcTemplate;

    // 状态数据库的JDBC模板（用于记录同步状态）----复用targetJdbcTemplate
    @Autowired
    @Qualifier("statusJdbcTemplate")
    private JdbcTemplate statusJdbcTemplate;

    // 需要同步的表集合（格式：库名.表名）
    private Set<String> SYNC_TABLE_SET = new HashSet<>();
    // 表的同步状态缓存（表名 -> 是否已全量同步）
    private Map<String, Boolean> syncStatusCache = new ConcurrentHashMap<>();
    // 同步状态表名称
    private static final String SYNC_STATUS_TABLE = "canal_sync_status";

    @PostConstruct
    public void init() {
        // 初始化同步的表集合
        initSyncTableSet();
        // 初始化同步状态表
        initSyncStatusTable();
        // 加载同步状态到缓存
        loadSyncStatus();
        //同步前先判断是否清除数据
        deleteTargetData();
        // 启动同步流程
        startSyncProcess();
    }

    //第一步：初始化同步的表集合
    private void initSyncTableSet() {
        //获取数据库名称
        String dbName = canalProperties.getDbName();
        if (!StrUtil.isEmpty(dbName)) {
            Set<String> tableNames = canalProperties.getTableSet();
            if (!BeanUtil.isEmpty(tableNames)) {
                log.info("已从配置中读取到了-->DB名称：{}, 表名称：{}", dbName, tableNames);

                // 如果表名集合包含"*"，则查询数据库中的所有基础表
                if (tableNames.contains("*")) {
                    try {
                        // 查询数据库中的所有基础表
                        String sql = "SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = ? AND TABLE_TYPE = 'BASE TABLE'";
                        List<Map<String, Object>> result = sourceJdbcTemplate.queryForList(sql, dbName);

                        if (CollectionUtil.isNotEmpty(result)) {
                            // 创建一个新的表名集合，替换"*"为所有表名
                            Set<String> newTableNames = new HashSet<>();
                            for (Map<String, Object> row : result) {
                                newTableNames.add((String) row.get("TABLE_NAME"));
                            }
                            tableNames = newTableNames;
                            log.info("已将'*'替换为数据库{}中的所有基础表，共{}个表", dbName, tableNames.size());
                        }
                    } catch (Exception e) {
                        log.error("查询数据库{}中的表失败: {}", dbName, e.getMessage(), e);
                    }
                }

                // 构建完整的表名（库名.表名）
                for (String tableName : tableNames) {
                    this.SYNC_TABLE_SET.add(dbName + "." + tableName);
                }
                log.info("已从配置中加载需要同步的表集合: {}", SYNC_TABLE_SET);

            } else {
                log.error("未配置表名称(tableName)，无法构建表集合");
            }
        } else {
            log.error("未配置数据库名称(dbName)，无法构建表集合");
        }
    }

    //第二步：初始化同步状态表
    private void initSyncStatusTable() {
        try {
            // 创建同步状态表（如果不存在）
            String createTableSql = String.format(
                    "CREATE TABLE IF NOT EXISTS %s ("
                            + "id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',"
                            + "db_name VARCHAR(64) NOT NULL COMMENT '源数据库名',"
                            + "table_name VARCHAR(64) NOT NULL COMMENT '源表名',"
                            + "is_full_sync TINYINT NOT NULL DEFAULT 0 COMMENT '是否已全量同步：0-未全量，1-已全量',"
                            + "full_sync_time DATETIME DEFAULT NULL COMMENT '全量同步完成时间',"
                            + "update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '状态更新时间',"
                            + "PRIMARY KEY (id),"
                            + "UNIQUE KEY uk_db_table (db_name,table_name) COMMENT '确保数据库+表唯一'"
                            + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT 'Canal 同步状态';",
                    SYNC_STATUS_TABLE
            );
            statusJdbcTemplate.update(createTableSql);
        } catch (Exception e) {
            log.error("同步状态表初始化失败: {}", e.getMessage(), e);
        }
    }

    //第三步：加载同步状态到缓存
    private void loadSyncStatus() {
        try {
            String sql = String.format("SELECT db_name, table_name FROM %s WHERE is_full_sync = 1", SYNC_STATUS_TABLE);
            List<Map<String, Object>> statusList = statusJdbcTemplate.queryForList(sql);
            for (Map<String, Object> status : statusList) {
                String dbName = (String) status.get("db_name");
                String tableName = (String) status.get("table_name");
                String fullTableName = dbName + "." + tableName;

                // 更新同步状态缓存
                syncStatusCache.put(fullTableName, true);
            }
            log.info("同步状态加载成功");
        } catch (Exception e) {
            log.error("同步状态加载失败: {}", e.getMessage(), e);
        }
    }

    //第四步：同步前先判断是否清除数据
    public void deleteTargetData() {
        try {
            // 检查是否是首次使用（没有记录的同步状态）
            String checkSql = "SELECT COUNT(*) FROM " + SYNC_STATUS_TABLE;
            Integer count = statusJdbcTemplate.queryForObject(checkSql, Integer.class);

            if (count < SYNC_TABLE_SET.size() || syncStatusCache.size() != SYNC_TABLE_SET.size()) {
                log.info("清理不存在状态库的表数据");

                //删除目标数据库数据--为了后面的全量同步数据准确性
                for (String fullTableName : SYNC_TABLE_SET) {
                    if (!syncStatusCache.containsKey(fullTableName)) {
                        try {
                            // 截取表名
                            String dbName = fullTableName.substring(0, fullTableName.lastIndexOf('.'));
                            String tableName = fullTableName.substring(fullTableName.lastIndexOf('.') + 1);

                            // 清空表数据
//                        String truncateSql = "TRUNCATE TABLE `" + tableName + "`";
                            String truncateSql = "DELETE FROM `" + tableName + "`";
                            targetJdbcTemplate.execute(truncateSql);
                            log.info("已清空目标数据库表：{}", tableName);

                            String deleteSql = "DELETE FROM " + SYNC_STATUS_TABLE + " WHERE db_name = ? AND table_name = ?";
                            statusJdbcTemplate.update(deleteSql, dbName, tableName);
                        } catch (Exception e) {
                            log.error("清空表数据失败 - 表名: {}, 错误信息: {}", fullTableName, e.getMessage());
                            // 继续处理下一个表，而不是中断整个过程
                        }
                    }
                }
                log.info("目标数据库数据清理完成");
            } else {
                log.info("非首次使用，跳过目标数据库数据清理步骤");
            }
        } catch (Exception e) {
            log.error("目标数据库数据清理过程中发生异常: {}", e.getMessage(), e);
        }
    }

    //第五步：启动同步流程
    private void startSyncProcess() {
        try {

            // 启动全量同步
            startFullSync();

            // 连接Canal
//            connectCanal();

            // 启动增量同步线程
//            startIncrementSync();

        } catch (Exception e) {
            log.error("同步流程启动失败: {}", e.getMessage(), e);
        }
    }

    //第六步：启动全量同步
    private void startFullSync() {
        for (String fullTableName : SYNC_TABLE_SET) {
            String[] parts = fullTableName.split("\\.");
            if (parts.length != 2) {
                log.error("无效的表名格式: {}", fullTableName);
                continue;
            }
            String dbName = parts[0];
            String tableName = parts[1];

            if (tableName.equals(SYNC_STATUS_TABLE))
                continue;

            // 检查是否已全量同步
            if (!syncStatusCache.getOrDefault(fullTableName, false)) {

                // 全量同步数据
                fullSyncData(dbName, tableName);

                // 更新同步状态
                updateSyncStatus(dbName, tableName);

            } else {
                log.info("表{}已完成全量同步，跳过全量同步步骤", fullTableName);
            }
        }
    }

    //第七步：连接 Canal
    private void connectCanal() {
        int maxRetries = canalProperties.getMaxRetries() > 0 ? canalProperties.getMaxRetries() : 5;
        int retryInterval = canalProperties.getRetryInterval() > 0 ? canalProperties.getRetryInterval() : 5000;
        int retryCount = 0;

        while (true) {
            try {
                // 先断开已有的连接
                try {
                    canalConnector.disconnect();
                } catch (Exception e) {
                    // 忽略断开连接的异常
                }

                // 重新连接Canal
                canalConnector.connect();
                canalConnector.subscribe(canalProperties.getSubscribe());
                log.info("Canal连接成功");
                return; // 连接成功，退出循环
            } catch (Exception e) {
                retryCount++;
                log.error("Canal连接失败 (尝试 {}/{}): {}", retryCount, maxRetries, e.getMessage());

                if (retryCount >= maxRetries) {
                    log.error("达到最大重试次数，放弃连接");
                    throw new RuntimeException("Canal连接失败", e);
                }

                try {
                    // 等待一段时间后重试
                    Thread.sleep(retryInterval);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    log.error("线程被中断，放弃重试");
                    throw new RuntimeException("Canal连接中断", ie);
                }
            }
        }
    }

    //第八步：启动增量同步
    private void startIncrementSync() {
        // 创建一个新线程来处理增量同步
        new Thread(() -> {
            boolean isRunning = true;

            while (isRunning) {
                try {
                    // 确保Canal连接已建立
                    if (!isCanalConnected()) {
                        log.info("尝试重新连接Canal...");
                        connectCanal();
                        // 重新设置消费位点
                        resetConsumePosition();
                    }

                    // 循环消费增量变更
                    while (true) {
                        try {
                            // 每次拉取1000条变更
                            Message message = canalConnector.getWithoutAck(1000);
                            long batchId = message.getId();
                            int size = message.getEntries().size();

                            if (batchId == -1 || size == 0) {
                                // 无变更，休眠1秒再拉取
                                Thread.sleep(1000);
                                continue;
                            }

                            // 处理增量变更数据并确认消费
                            try {
                                handleIncrementData(message.getEntries());
                                canalConnector.ack(batchId);
                                log.info("本批次[{}]增量同步完成，共处理{}条变更", batchId, size);
                            } catch (Exception e) {
                                // 处理数据异常时，尝试回滚该批次
                                handleBatchException(batchId, e);
                            }
                        } catch (Exception e) {
                            // 处理连接异常或其他同步异常
                            if (handleSyncException(e)) {
                                break; // 跳出内部循环，触发重连
                            }
                            // 非连接异常，休眠后继续
                            Thread.sleep(3000);
                        }
                    }
                } catch (Exception e) {
                    log.error("增量同步线程异常: {}", e.getMessage(), e);

                    try {
                        // 发生严重异常，休眠一段时间后再尝试
                        Thread.sleep(5000);
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        isRunning = false;
                        log.error("线程被中断，停止增量同步");
                    }
                }
            }
        }, "canal-increment-sync-thread").start();
    }


    // 全量同步核心步骤：

    /**
     * 执行全量同步
     */
    private void fullSyncData(String dbName, String tableName) {
        try {

            // 检查表是否存在，如果不存在则创建表
            if (!checkTableExists(tableName)) {
                log.info("目标数据库中表 {} 不存在，开始创建表", tableName);
                createTableFromSource(dbName, tableName);
            }

            // 执行全量同步--获取当前时间并格式化
            LocalDateTime now = LocalDateTime.now();
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSS");
            String formattedStartTime = now.format(formatter);
            log.info("开始全量同步表: {}, 当前时间：{}", tableName, formattedStartTime);

            List<String> columnNames = getTableColumns(dbName, tableName);

            if (CollectionUtil.isEmpty(columnNames)) {
                log.error("获取表结构失败，取消全量同步");
            }

            int pageSize = 20000;
            long totalRows = 0;
            Object lastPrimaryKeyValue = null; // 上次查询的最大主键值

            // 获取表的主键
            String primaryKey = getTablePrimaryKey(dbName, tableName);
            if (StrUtil.isEmpty(primaryKey)) {
                primaryKey = "id"; // 默认主键为id
                log.warn("无法获取表主键，使用默认值: id");
            }

            while (true) {
                // 基于主键值的分页查询SQL
                String sourceSql;
                List<Object> params = new ArrayList<>();

                // 处理列名，为MySQL保留字添加反引号转义
                List<String> escapedColumnNames = columnNames.stream()
                        .map(column -> "`" + column + "`")
                        .collect(Collectors.toList());

                // 对主键也进行转义
                String escapedPrimaryKey = "`" + primaryKey + "`";

                if (lastPrimaryKeyValue == null) {
                    // 首次查询，没有上次的主键值
                    sourceSql = String.format(
                            "SELECT %s FROM %s.%s ORDER BY %s LIMIT ?",
                            String.join(",", escapedColumnNames),
                            dbName,
                            tableName,
                            escapedPrimaryKey
                    );
                    params.add(pageSize);
                } else {
                    // 非首次查询，使用上次查询的最大主键值作为条件
                    sourceSql = String.format(
                            "SELECT %s FROM %s.%s WHERE %s > ? ORDER BY %s LIMIT ?",
                            String.join(",", escapedColumnNames),
                            dbName,
                            tableName,
                            escapedPrimaryKey,
                            escapedPrimaryKey
                    );
                    params.add(lastPrimaryKeyValue);
                    params.add(pageSize);
                }

                List<Map<String, Object>> pageData = sourceJdbcTemplate.queryForList(sourceSql, params.toArray());
                if (CollectionUtil.isEmpty(pageData)) {
                    break; // 没有更多数据了
                }

                // 写入目标数据库
                batchInsertData(tableName, columnNames, pageData);

                // 更新计数器
                totalRows += pageData.size();

                // 更新上次查询的最大主键值
                Map<String, Object> lastRow = pageData.get(pageData.size() - 1);
                lastPrimaryKeyValue = lastRow.get(primaryKey);
                if (lastPrimaryKeyValue == null) {
                    //其实这里仍然没有解决关于没有主键，没有id字段，或者是有id字段，但是id字段值为null的bug
                    throw new InvalidDataAccessResourceUsageException("表" + tableName + "一共同步了" + totalRows + "条数据");
                }
            }

            // 计算耗时并格式化完成时间
            LocalDateTime endTime = LocalDateTime.now();
            long durationInMillis = Duration.between(now, endTime).toMillis();

            log.info("表{}.{}全量同步完成，共同步{}条数据，总耗时：{}ms", dbName, tableName, totalRows, durationInMillis);
        } catch (InvalidDataAccessResourceUsageException e) {
            log.error("该表{}没有主键且没有id字段, 全量同步失败, {}", tableName, e.getMessage());
        } catch (Exception e) {
            log.error("全量同步失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 批量插入数据到目标表
     */
    private void batchInsertData(String tableName, List<String> columnNames, List<Map<String, Object>> dataList) {
        try {
            if (CollectionUtil.isEmpty(dataList)) {
                return;
            }

            // 构建INSERT语句 - 对表名和列名添加反引号转义以处理MySQL保留字
            String escapedTableName = "`" + tableName + "`";
            List<String> escapedColumnNames = columnNames.stream()
                    .map(column -> "`" + column + "`")
                    .collect(Collectors.toList());
            String columns = String.join(",", escapedColumnNames);
            String placeholders = String.join(",", Collections.nCopies(columnNames.size(), "?"));
            String insertSql = String.format("INSERT INTO %s (%s) VALUES (%s)", escapedTableName, columns, placeholders);

            // 准备批处理参数
            List<Object[]> batchArgs = new ArrayList<>(dataList.size());
            for (Map<String, Object> row : dataList) {
                Object[] args = new Object[columnNames.size()];
                for (int i = 0; i < columnNames.size(); i++) {
                    String columnName = columnNames.get(i);
                    args[i] = row.get(columnName);
                }
                batchArgs.add(args);
            }

            // 记录开始时间，用于性能监控
            long startTime = System.currentTimeMillis();

            // 批量执行SQL - 直接使用更高效的batchUpdate重载方法
            int[] updateCounts = targetJdbcTemplate.batchUpdate(insertSql, batchArgs);

            // 记录结束时间和性能统计
            long endTime = System.currentTimeMillis();
            long duration = endTime - startTime;
            // 计算平均每条记录耗时（确保结果是double类型）
            double avgTimePerRecord = updateCounts.length > 0 ? (double) duration / updateCounts.length : 0.0;
            log.debug("批量插入表{}数据: {}条, 耗时: {}ms, 平均每条: {}ms",
                    tableName, updateCounts.length, duration, avgTimePerRecord);

            // 检查是否有失败的插入
            for (int i = 0; i < updateCounts.length; i++) {
                if (updateCounts[i] == Statement.EXECUTE_FAILED) {
                    log.warn("批量插入表{}时，第{}条数据插入失败", tableName, i + 1);
                }
            }
        } catch (Exception e) {
            log.error("批量插入表{}数据失败: {}", tableName, e.getMessage(), e);
        }
    }

    /**
     * 更新同步状态
     */
    private void updateSyncStatus(String dbName, String tableName) {
        try {
            String fullTableName = dbName + "." + tableName;

            // 更新数据库状态
            String insertSql = String.format(
                    "INSERT INTO %s (db_name, table_name, is_full_sync, full_sync_time) VALUES (?, ?, ?, NOW())",
                    SYNC_STATUS_TABLE
            );
            statusJdbcTemplate.update(insertSql, dbName, tableName, 1);

            // 更新缓存状态
            syncStatusCache.put(fullTableName, true);
            log.info("{}更新状态成功", fullTableName);
        } catch (Exception e) {
            log.error("更新同步状态失败: {}", e.getMessage(), e);
        }
    }


    // 增量同步核心步骤：

    /**
     * 处理增量变更数据
     */
    private void handleIncrementData(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            // 过滤非行数据变更（如事务开始/结束）
            if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                continue;
            }

            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("解析binlog变更失败: {}", e.getMessage(), e);
                continue;
            }

            // 获取数据库名、表名和当前binlog位点信息
            String dbName = entry.getHeader().getSchemaName();
            String tableName = entry.getHeader().getTableName();
            String fullTableName = dbName + "." + tableName;

            // 只处理需要同步的表
            if (!SYNC_TABLE_SET.contains(fullTableName)) {
                continue;
            }


            CanalEntry.EventType eventType = rowChange.getEventType();
            try {
                // 先输出基本信息日志
                log.info("处理增量变更: db={}, table={}, type={}, 数据条数={}",
                        dbName, tableName, eventType, rowChange.getRowDatasList().size());

                for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                    StringBuilder changeDetail = new StringBuilder();

                    // 根据事件类型显示不同的变更信息
                    if (eventType == CanalEntry.EventType.INSERT) {
                        // INSERT事件只显示新增的数据
                        changeDetail.append("插入数据: ").append(formatColumns(rowData.getAfterColumnsList()));
                    } else if (eventType == CanalEntry.EventType.UPDATE) {
                        // UPDATE事件显示变更前后的数据
                        changeDetail.append("更新前: ").append(formatColumns(rowData.getBeforeColumnsList()))
                                .append(", " + "\n" + "更新后: ").append(formatColumns(rowData.getAfterColumnsList()));
                    } else if (eventType == CanalEntry.EventType.DELETE) {
                        // DELETE事件只显示删除的数据
                        changeDetail.append("删除数据: ").append(formatColumns(rowData.getBeforeColumnsList()));
                    }

                    log.info("数据变更详情: {}", changeDetail);
                }
            } catch (Exception e) {
                log.error("处理增量变更数据异常: {}", e.getMessage());
            }

            // 遍历每行变更数据，写入目标存储
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                switch (eventType) {
                    case INSERT:
                        handleInsert(tableName, rowData.getAfterColumnsList());
                        break;
                    case UPDATE:
                        handleUpdate(tableName, rowData.getBeforeColumnsList(), rowData.getAfterColumnsList());
                        break;
                    case DELETE:
                        handleDelete(tableName, rowData.getBeforeColumnsList());
                        break;
                    default:
                        log.warn("不支持的变更类型: {}", eventType);
                }
            }
        }
    }

    /**
     * 处理INSERT操作
     * 使用INSERT INTO ... ON DUPLICATE KEY UPDATE来避免全量同步后增量同步的主键冲突问题
     */
    private void handleInsert(String tableName, List<CanalEntry.Column> afterColumns) {

        try {

            // 提取字段名和值
            List<String> columns = new ArrayList<>();
            List<Object> values = new ArrayList<>();
            StringBuilder dataInfo = new StringBuilder();
            List<String> updateClauses = new ArrayList<>();

            for (CanalEntry.Column column : afterColumns) {
                String columnName = column.getName();
                columns.add("`" + columnName + "`");
                Object value = getColumnValue(column);
                values.add(value);

                // 构建ON DUPLICATE KEY UPDATE子句中的更新表达式
                // 注意：主键列不需要包含在更新子句中
                updateClauses.add("`" + columnName + "` = VALUES(`" + columnName + "`)");

                if (dataInfo.length() > 0) {
                    dataInfo.append(", ");
                }
                dataInfo.append(columnName).append("='").append(value).append("'");
            }

            // 生成INSERT SQL，使用ON DUPLICATE KEY UPDATE避免主键冲突
            String sql = String.format(
                    "INSERT INTO %s (%s) VALUES (%s) ON DUPLICATE KEY UPDATE %s",
                    tableName,
                    String.join(",", columns),
                    String.join(",", Collections.nCopies(columns.size(), "?")),
                    String.join(", ", updateClauses)
            );

            // 执行SQL
            int affectedRows = targetJdbcTemplate.update(sql, values.toArray());

            // 记录操作结果
            if (affectedRows > 0) {
                // affectedRows > 1表示发生了更新，而不是插入
                if (affectedRows > 1) {
                    log.debug("处理INSERT操作 - 表名: {}, 数据已存在并更新", tableName);
                } else {
                    log.debug("处理INSERT操作 - 表名: {}, 成功插入新数据", tableName);
                }
            }
        } catch (Exception e) {
            // 记录更详细的失败信息
            StringBuilder errorDataInfo = new StringBuilder();
            for (CanalEntry.Column column : afterColumns) {
                if (errorDataInfo.length() > 0) {
                    errorDataInfo.append(", ");
                }
                errorDataInfo.append(column.getName()).append("='").append(column.getValue()).append("'");
            }
            log.error("处理INSERT操作失败 - 表名: {}, 数据: {}, 错误原因: {}, 解释：{}",
                    tableName, errorDataInfo, e.getMessage(), "可能该条数据已经存在");
        }
    }

    /**
     * 处理UPDATE操作
     */
    private void handleUpdate(String tableName, List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns) {
        try {
            // 提取更新的字段和新值
            List<String> setClauses = new ArrayList<>();
            List<Object> setValues = new ArrayList<>();
            // 提取主键（假设表的主键是"id"，若不是需修改）
            String primaryKey = "id";
            Object primaryValue = null;

            // 从beforeColumns中获取主键值
            for (CanalEntry.Column beforeCol : beforeColumns) {
                if (beforeCol.getName().equals(primaryKey)) {
                    primaryValue = getColumnValue(beforeCol);
                    break;
                }
            }

            if (primaryValue == null) {
                log.error("无法获取主键值，跳过UPDATE操作 - 表名: {}", tableName);
                return;
            }

            // 生成SET子句和更新字段信息
            StringBuilder updateFieldsInfo = new StringBuilder();
            for (CanalEntry.Column afterCol : afterColumns) {
                // 跳过主键列
                if (afterCol.getName().equals(primaryKey)) {
                    continue;
                }
                setClauses.add("`" + afterCol.getName() + "` = ?");
                setValues.add(getColumnValue(afterCol));
                if (updateFieldsInfo.length() > 0) {
                    updateFieldsInfo.append(", ");
                }
                updateFieldsInfo.append(afterCol.getName()).append("='").append(afterCol.getValue()).append("'");
            }

            // 添加主键到参数列表
            setValues.add(primaryValue);

            // 生成UPDATE SQL
            String sql = String.format(
                    "UPDATE %s SET %s WHERE %s = ?",
                    tableName,
                    String.join(", ", setClauses),
                    primaryKey
            );

            // 执行SQL
            targetJdbcTemplate.update(sql, setValues.toArray());
        } catch (Exception e) {
            // 记录更详细的失败信息
            String primaryValueStr = null;
            // 尝试获取主键值
            for (CanalEntry.Column beforeCol : beforeColumns) {
                if (beforeCol.getName().equals("id")) {
                    primaryValueStr = beforeCol.getValue();
                    break;
                }
            }
            // 构建更新字段信息
            StringBuilder updateFieldsInfo = new StringBuilder();
            for (CanalEntry.Column afterCol : afterColumns) {
                if (!afterCol.getName().equals("id")) {
                    if (updateFieldsInfo.length() > 0) {
                        updateFieldsInfo.append(", ");
                    }
                    updateFieldsInfo.append(afterCol.getName()).append("='").append(afterCol.getValue()).append("'");
                }
            }
            log.error("处理UPDATE操作失败 - 表名: {}, 主键: {}={}, 更新字段: {}, 错误原因: {}, 解释：{}",
                    tableName, "id", primaryValueStr, updateFieldsInfo, e.getMessage(), "可能该条数据已经变更");
        }
    }

    /**
     * 处理DELETE操作
     */
    private void handleDelete(String tableName, List<CanalEntry.Column> beforeColumns) {
        try {
            // 提取主键（假设主键是"id"，若不是需修改）
            String primaryKey = "id";
            Object primaryValue = null;

            for (CanalEntry.Column column : beforeColumns) {
                if (column.getName().equals(primaryKey)) {
                    primaryValue = getColumnValue(column);
                    break;
                }
            }

            if (primaryValue == null) {
                log.error("无法获取主键值，跳过DELETE操作 - 表名: {}", tableName);
                return;
            }

            // 生成DELETE SQL
            String sql = String.format(
                    "DELETE FROM %s WHERE %s = ?",
                    tableName,
                    primaryKey
            );

            // 执行SQL
            targetJdbcTemplate.update(sql, primaryValue);
        } catch (Exception e) {
            // 记录更详细的失败信息
            String primaryValueStr = null;
            // 尝试获取主键值
            for (CanalEntry.Column column : beforeColumns) {
                if (column.getName().equals("id")) {
                    primaryValueStr = column.getValue();
                    break;
                }
            }
            // 构建完整的数据信息
            StringBuilder dataInfo = new StringBuilder();
            for (CanalEntry.Column column : beforeColumns) {
                if (dataInfo.length() > 0) {
                    dataInfo.append(", ");
                }
                dataInfo.append(column.getName()).append("='").append(column.getValue()).append("'");
            }
            log.error("处理DELETE操作失败 - 表名: {}, 主键: {}={}, 完整数据: {}, 错误原因: {}, 解释：{}",
                    tableName, "id", primaryValueStr, dataInfo, e.getMessage(), "可能该条数据已经删除");
        }
    }

    /**
     * 重置消费位置
     */
    private void resetConsumePosition() {
        try {
            // 设置消费位点
            log.info("设置从当前位置开始消费");
            // 先rollback清理之前的状态
            canalConnector.rollback();

            try {
                // 尝试直接设置订阅位置
                canalConnector.subscribe(canalProperties.getSubscribe());
            } catch (Exception e) {
                log.warn("重新订阅时发生异常，尝试备选方案: {}", e.getMessage());
                // 备选方案：先取消订阅再重新订阅
                try {
                    canalConnector.unsubscribe();
                    Thread.sleep(500); // 短暂休眠避免频繁操作
                    canalConnector.subscribe(canalProperties.getSubscribe());
                } catch (Exception ex) {
                    log.error("备选方案也失败: {}", ex.getMessage());
                }
            }
        } catch (Exception e) {
            log.error("重置消费位置失败: {}", e.getMessage(), e);
        }
    }


    // 其他方法

    /**
     * 获取表的所有列名
     */
    private List<String> getTableColumns(String dbName, String tableName) {
        try {
            String sql = String.format(
                    "SELECT COLUMN_NAME FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s'",
                    dbName,
                    tableName
            );
            return sourceJdbcTemplate.queryForList(sql, String.class);
        } catch (Exception e) {
            log.error("获取表结构失败: {}", e.getMessage(), e);
            return new ArrayList<>();
        }
    }

    /**
     * 获取表的主键
     */
    private String getTablePrimaryKey(String dbName, String tableName) {
        try {
            String sql = String.format(
                    "SELECT COLUMN_NAME FROM information_schema.KEY_COLUMN_USAGE WHERE TABLE_SCHEMA = '%s' AND TABLE_NAME = '%s' AND CONSTRAINT_NAME = 'PRIMARY'",
                    dbName,
                    tableName
            );
            List<String> keys = sourceJdbcTemplate.queryForList(sql, String.class);
            return CollectionUtil.isEmpty(keys) ? null : keys.get(0);
        } catch (Exception e) {
            log.error("获取表主键失败: {}", e.getMessage(), e);
            return null;
        }
    }

    /**
     * 检查Canal连接状态
     */
    private boolean isCanalConnected() {
        try {
            // 尝试一个轻量级操作来验证连接状态
            canalConnector.getWithoutAck(0);
            return true;
        } catch (Exception e) {
            log.error("Canal连接已断开: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 处理批次异常
     */
    private void handleBatchException(long batchId, Exception e) {
        try {
            canalConnector.rollback(batchId);
            log.warn("回滚批次[{}]数据处理异常: {}", batchId, e.getMessage());
        } catch (Exception re) {
            log.error("回滚批次[{}]失败: {}", batchId, re.getMessage());
        }
        // 休眠一段时间再继续
        try {
            Thread.sleep(3000);
        } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
        }
    }

    /**
     * 处理同步异常
     *
     * @return 是否需要触发重连
     */
    private boolean handleSyncException(Exception e) {
        log.error("增量同步处理异常: {}", e.getMessage(), e);

        // 检查是否是连接相关的异常或批次ID异常
        if (isConnectionException(e)) {
            log.warn("检测到连接异常、反序列化失败或批次ID异常，将尝试重新连接");
            // 尝试清空可能的错误批次
            try {
                canalConnector.rollback();
            } catch (Exception re) {
                // 忽略回滚失败，因为我们本来就要重连
            }
            return true;
        }

        // 非连接异常，回滚批次
        try {
            canalConnector.rollback();
        } catch (Exception re) {
            log.error("回滚批次失败: {}", re.getMessage());
        }

        return false;
    }

    /**
     * 检查异常是否与连接相关或批次ID异常
     */
    private boolean isConnectionException(Exception e) {
        String message = e.getMessage();
        return message != null && (
                message.contains("Connection refused") ||
                        message.contains("Connection reset") ||
                        message.contains("Broken pipe") ||
                        message.contains("connection is closed") ||
                        message.contains("channel is inactive") ||
                        // 添加对deserializer failed和batchId异常的检测
                        e instanceof com.alibaba.otter.canal.protocol.exception.CanalClientException && (
                                message.contains("deserializer failed") ||
                                        message.contains("batchId") ||
                                        message.contains("ack error")
                        )
        );
    }


    /**
     * 格式化列数据为可读的字符串
     */
    private String formatColumns(List<CanalEntry.Column> columns) {
        StringBuilder result = new StringBuilder();
        result.append("{");
        boolean first = true;
        for (CanalEntry.Column column : columns) {
            if (!first) {
                result.append(", ");
            }
            column.getValue();
            result.append(column.getName()).append("='").append(column.getValue()).append("'");
            first = false;
        }
        result.append("}");
        return result.toString();
    }


    /**
     * 获取列的实际值
     */
    private Object getColumnValue(CanalEntry.Column column) {
        if (column.getIsNull()) {
            return null;
        }
        // 根据列类型进行适当的转换（这里简化处理）
        return column.getValue();
    }

    /**
     * 检查目标数据库中表是否存在
     */
    private boolean checkTableExists(String tableName) {
        try {
            String sql = "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ?";
            Integer count = targetJdbcTemplate.queryForObject(sql, Integer.class, tableName);
            return count != null && count > 0;
        } catch (Exception e) {
            log.error("检查表是否存在失败: {}", e.getMessage());
            return false;
        }
    }

    /**
     * 从源数据库复制表结构并在目标数据库中创建表
     */
    private void createTableFromSource(String dbName, String tableName) {
        try {
            // 从源数据库获取建表语句
            String showCreateTableSql = String.format("SHOW CREATE TABLE %s.%s", dbName, tableName);
            Map<String, Object> result = sourceJdbcTemplate.queryForMap(showCreateTableSql);
            String createTableSql = (String) result.get("Create Table");

            // 在目标数据库中执行建表语句
            targetJdbcTemplate.execute(createTableSql);
            log.info("成功在目标数据库创建表: {}", tableName);
        } catch (Exception e) {
            log.error("创建表失败 - 表名: {}, 错误原因: {}", tableName, e.getMessage(), e);
        }
    }

}